package com.central.scheduler.master.builder;

import com.central.scheduler.master.runner.task.ITaskProcessFactory;
import com.central.scheduler.spi.task.request.DataxTaskExecutionContext;
import com.central.scheduler.common.enums.TaskTimeoutStrategy;
import com.central.scheduler.common.enums.TimeoutFlag;
import com.central.scheduler.dao.model.ProcessDefinition;
import com.central.scheduler.dao.model.ProcessInstance;
import com.central.scheduler.dao.model.TaskInstance;
import com.central.scheduler.service.queue.entity.TaskExecutionContext;
import com.central.scheduler.spi.task.request.FlinkSqlTaskExecutionContext;
import com.central.scheduler.spi.task.request.SQLTaskExecutionContext;

import static com.central.scheduler.common.Constants.SEC_2_MINUTES_TIME_UNIT;

/**
 * @author Tindy
 * @date 2021/11/3
 * @describe
 *  TaskExecutionContext builder
 */
public class TaskExecutionContextBuilder {
    public static TaskExecutionContextBuilder get() {
        return new TaskExecutionContextBuilder();
    }
    private TaskExecutionContext taskExecutionContext =  new TaskExecutionContext();

    /**
     * build taskInstance related info
     *
     * @param taskInstance taskInstance
     * @return TaskExecutionContextBuilder
     */
    public TaskExecutionContextBuilder buildTaskInstanceRelatedInfo(TaskInstance taskInstance) {
        taskExecutionContext.setTaskInstanceId(taskInstance.getId());
        taskExecutionContext.setTaskName(taskInstance.getName());
        taskExecutionContext.setFirstSubmitTime(taskInstance.getFirstSubmitTime());
        taskExecutionContext.setStartTime(taskInstance.getStartTime());
        taskExecutionContext.setTaskType(taskInstance.getTaskType());
        taskExecutionContext.setLogPath(taskInstance.getLogPath());
        taskExecutionContext.setWorkerGroupId(taskInstance.getWorkerGroupId());
        taskExecutionContext.setEnvironmentConfig(taskInstance.getEnvironmentConfig());
        taskExecutionContext.setHost(taskInstance.getHost());
        taskExecutionContext.setResources(taskInstance.getResources());
        taskExecutionContext.setDelayTime(taskInstance.getDelayTime());
        taskExecutionContext.setLocalParams(taskInstance.getLocalParams());
        taskExecutionContext.setGlobalParams(taskInstance.getLocalParams());
        taskExecutionContext.setVarPool(taskInstance.getVarPool());
        taskExecutionContext.setDryRun(taskInstance.getDryRun());
        taskExecutionContext.setTaskTimeout(Integer.MAX_VALUE);
        if (taskInstance.getTimeoutParam().getFlag() == TimeoutFlag.OPEN) {
            taskExecutionContext.setTaskTimeoutStrategy(taskInstance.getTimeoutParam().getStrategy());
            if (taskInstance.getTimeoutParam().getStrategy() == TaskTimeoutStrategy.FAILED
                    || taskInstance.getTimeoutParam().getStrategy() == TaskTimeoutStrategy.WARNFAILED) {
                taskExecutionContext.setTaskTimeout(Math.min(taskInstance.getTimeoutParam().getInterval() * SEC_2_MINUTES_TIME_UNIT, Integer.MAX_VALUE));
            }
        }
        taskExecutionContext.setTaskContent(taskInstance.getTaskContent());
        return this;
    }

    /**
     * build processInstance related info
     *
     * @param processInstance processInstance
     * @return TaskExecutionContextBuilder
     */
    public TaskExecutionContextBuilder buildProcessInstanceRelatedInfo(ProcessInstance processInstance) {
        taskExecutionContext.setProcessInstanceId(processInstance.getId());
        taskExecutionContext.setScheduleTime(processInstance.getScheduleTime());
        taskExecutionContext.setGlobalParams(processInstance.getGlobalParams());
        taskExecutionContext.setUserId(processInstance.getUserId());
        taskExecutionContext.setUsername(processInstance.getUsername());
        taskExecutionContext.setProId(processInstance.getProId());
        taskExecutionContext.setProName(processInstance.getProName());
        taskExecutionContext.setCmdTypeIfComplement(processInstance.getCmdTypeIfComplement().getCode());
        return this;
    }
    /**
     * build processDefinition related info
     *
     * @param processDefinition processDefinition
     * @return TaskExecutionContextBuilder
     */
    public TaskExecutionContextBuilder buildProcessDefinitionRelatedInfo(ProcessDefinition processDefinition) {
        taskExecutionContext.setProcessDefineCode(processDefinition.getCode());
        taskExecutionContext.setProcessDefineVersion(processDefinition.getVersion());
        taskExecutionContext.setProId(processDefinition.getProId());
        return this;
    }

    /**
     * build SQLTask related info
     *
     * @param sqlTaskExecutionContext sqlTaskExecutionContext
     * @return TaskExecutionContextBuilder
     */
    public TaskExecutionContextBuilder buildSQLTaskRelatedInfo(SQLTaskExecutionContext sqlTaskExecutionContext) {
        taskExecutionContext.setSqlTaskExecutionContext(sqlTaskExecutionContext);
        return this;
    }

    /**
     * build DataxTask related info
     *
     * @param dataxTaskExecutionContext dataxTaskExecutionContext
     * @return TaskExecutionContextBuilder
     */
    public TaskExecutionContextBuilder buildDataxTaskRelatedInfo(DataxTaskExecutionContext dataxTaskExecutionContext) {
        taskExecutionContext.setDataxTaskExecutionContext(dataxTaskExecutionContext);
        return this;
    }

    /**
     * create
     *
     * @return taskExecutionContext
     */
    public TaskExecutionContext create() {
        return taskExecutionContext;
    }

    public TaskExecutionContextBuilder buildFlinkSqlTaskRelatedInfo(FlinkSqlTaskExecutionContext flinkSqlTaskExecutionContext) {
        taskExecutionContext.setFlinkSqlTaskExecutionContext(flinkSqlTaskExecutionContext);
        return this;
    }
}
